package com.cloud.message.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cloud.common.bean.PageInfo;
import com.cloud.message.cache.ConsumerCache;
import com.cloud.message.dao.MessagePendingDao;
import com.cloud.message.dto.PendingQueryParamDTO;
import com.cloud.message.entity.Consumer;
import com.cloud.message.entity.MessagePending;
import com.cloud.message.rocketmq.MQMessage;
import com.cloud.message.rocketmq.RocketMQSender;
import com.cloud.message.service.MessagePendingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;

@Slf4j
@Service
public class MessagePendingServiceImpl extends ServiceImpl<MessagePendingDao, MessagePending> implements MessagePendingService {

    @Autowired
    private RocketMQSender rocketMQSender;

    @Autowired
    private ConsumerCache consumerCache;

    @Override
    public void createPending(String serviceName, MQMessage message) {
        MessagePending pending = new MessagePending();
        pending.setLevel(message.getLevel());
        pending.setTopic(message.getTopic());
        pending.setContent(message.getContent());
        pending.setHashKey(message.getHashKey());
        pending.setServiceName(serviceName);
        pending.setRetryCount(0);
        pending.setUserVersion(message.getUserVersion());
        pending.setUserSession(message.getUserSession());
        pending.setCreateTime(new Date());
        pending.setUpdateTime(pending.getCreateTime());
        save(pending);
    }

    @Override
    public void removePending(String serviceName, String topic) {
        QueryWrapper<MessagePending> wrapper = new QueryWrapper();
        wrapper.eq(MessagePending.SERVICE_NAME, serviceName);
        wrapper.eq(MessagePending.TOPIC, topic);
        remove(wrapper);
    }

    @Override
    public void handleRetry() {
        long beginId = 0;
        long pageSize = 1000;
        long currentTime = System.currentTimeMillis();
        while (true) {
            QueryWrapper<MessagePending> wrapper = new QueryWrapper();
            wrapper.gt(MessagePending.ID, beginId);
            List<MessagePending> list = page(new Page(1, pageSize), wrapper).getRecords();
            if (CollectionUtils.isEmpty(list)) {
                break;
            }
            for (MessagePending pending : list) {
                Consumer consumer = consumerCache.getConsumer(pending.getTopic(), pending.getServiceName());
                if (consumer == null) {
                    removeById(pending.getId());
                    continue;
                }
                if (pending.getRetryCount() > consumer.getRetryCount()) {
                    log.warn("message pending no processing, pendingId={}", pending.getId());
                    continue;
                }
                if (currentTime > pending.getUpdateTime().getTime() + 3000) {
                    rocketMQSender.sendPending(pending);
                }
            }
            if (list.size() < pageSize) {
                break;
            }
            beginId = CollectionUtils.lastElement(list).getId();
        }
    }

    @Override
    public PageInfo<MessagePending> getPage(PendingQueryParamDTO pendingQueryParamDTO) {
        QueryWrapper<MessagePending> wrapper = new QueryWrapper();
        IPage page = page(new Page(pendingQueryParamDTO.getPageNo(), pendingQueryParamDTO.getPageSize()), wrapper);
        return new PageInfo(page);
    }
}
